/**
 * 
 */
package com.hanhai.common.zookeeper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

/**
 * @author leiruiqi@hanhai.com
 *
 */

@Service("zookeeperLock")
public class ZookeeperLock{

	private static final Logger LOG = LoggerFactory.getLogger(ZookeeperLock.class); 
	
	@Resource
	private BaseZookeeper baseZookeeper;
	
	private String basePath="/locks";
	
	//private String lockPath = "/locks/testA";
	
	private String selfPath = "";
	
	  
	
	//创建连接
	@PostConstruct
	public void init() throws IOException, InterruptedException, KeeperException{
		this.creatPath(basePath, "");
		
	}
	
	public void creatPath(String path,String data) throws KeeperException, InterruptedException{
		if(baseZookeeper.zooKeeper.exists(path, false) == null){
			baseZookeeper.createNode(path, data.getBytes(), CreateMode.PERSISTENT);
			LOG.info("创建节点成功，path＝"+path);
		}else{
			LOG.info("创建节点失败，path＝"+path);
		}
		
	}
	
	public boolean simplelock(String childPath) {
		String value = null;
		boolean bool = false;
		try {
			value =	baseZookeeper.createNode(basePath+"/"+childPath, null, CreateMode.EPHEMERAL);
			LOG.info("simplelock result="+value);
			bool = true;
		} catch (KeeperException e) {
			// TODO Auto-generated catch block
			//e.printStackTrace();
			return false;
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return bool;	
	}
	
	public void simpleUnlock(String childPath){
		try {
			baseZookeeper.deletNode(basePath+"/"+childPath, -1);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (KeeperException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public void releaseConnection(){
		try {
			this.baseZookeeper.closeConnect();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public boolean lock(String childPath){
		String value = null;
		boolean bool = false;
		try {
			value =	baseZookeeper.createNode(basePath+"/"+childPath, null, CreateMode.EPHEMERAL_SEQUENTIAL);
			selfPath = value;
			
			List<String> subNodes = baseZookeeper.getChildren(basePath);
			
			Collections.sort(subNodes);
			int index = subNodes.indexOf( selfPath.substring(basePath.length()+1));  
			
			if(index==0){
				bool=true;
				
			}
			LOG.info("lock result="+bool);
			//LOG.info("selfPath="+selfPath+" ,small node="+smallNode);
			
		} catch (KeeperException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return bool;
	}
	
	public void unlock(String path){
		try {
			baseZookeeper.deletNode(selfPath, -1);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (KeeperException e) {
			// TODO Auto-generated catch block
			//e.printStackTrace();
		}
	}
	
	public static void main(String[] args) {

		String CONNECTION_STRING = "172.17.32.253:2181";
		final String GROUP_PATH = "/locks";
		//final String lockPath = "/locks/lockTestA";

		
		final String childPath = "localTestB";
		final int THREAD_NUM = 10; 
		//确保所有线程运行结束；  
	    final CountDownLatch threadSemaphore = new CountDownLatch(THREAD_NUM);  
	    

	    List<Thread> list = new ArrayList<Thread>();
	    
		 for(int i=0; i < THREAD_NUM; i++){  
	            final int threadId = i+1;  
	            try{  
                	BaseZookeeper bz = new BaseZookeeper();
                	bz.connectZookeeper(CONNECTION_STRING);
                	final ZookeeperLock zl = new ZookeeperLock();
                	zl.baseZookeeper = bz;
                	
                    synchronized (threadSemaphore){  
                        zl.creatPath(GROUP_PATH, "该节点由线程" + threadId + "创建");  
                    }  
                    
                    list.add(new Thread(){  
    	            	private ZookeeperLock z1 = zl;
    	            	private int tid  = threadId;
    	                @Override  
    	                public void run() {  
    	                    
							/*boolean locked = z1.lock(lockPath.substring(GROUP_PATH.length()+1));
							LOG.info(tid+" locked "+locked );
							if(locked){
								try {
									Thread.sleep(5000);
								} catch (InterruptedException e) {
									// TODO Auto-generated catch block
									e.printStackTrace();
								}
								
								LOG.info(tid+" unlocked "+lockPath);
							}
							z1.unlock(lockPath);
							*/
    	                	boolean locked = z1.simplelock(childPath);
    	                	LOG.info(tid+" locked "+locked );
    	                	if(locked){
								try {
									Thread.sleep(5000);
									LOG.info(tid+" unlocked "+childPath);
									
									List<String> clientList = z1.baseZookeeper.getChildren(GROUP_PATH);
									LOG.info(clientList.toString());
								} catch (InterruptedException e) {
									// TODO Auto-generated catch block
									e.printStackTrace();
								} catch (KeeperException e) {
									// TODO Auto-generated catch block
									e.printStackTrace();
								}
								
								z1.simpleUnlock(childPath);
								try {
									z1.creatPath(GROUP_PATH, "");
								 
									List<String> clientList = z1.baseZookeeper.getChildren(GROUP_PATH);
								
									LOG.info(clientList.toString());
								} catch (KeeperException | InterruptedException e) {
									// TODO Auto-generated catch block
									e.printStackTrace();
								}
								
							}
    	                	
    	                	
    	                	
							z1.releaseConnection();  
					        threadSemaphore.countDown(); 
    	                }  
    	            });  
                } catch (Exception e){  
                    LOG.error("【第"+threadId+"个线程】 抛出的异常：");  
                    e.printStackTrace();  
                }  
	            
	        }  
		 	for(Thread t:list){
		 		t.start();
		 	}
	        try {  
	            threadSemaphore.await();  
	            LOG.info("所有线程运行结束!");  
	        } catch (InterruptedException e) {  
	            e.printStackTrace();  
	        }  
	}
	
}
